﻿using LJ.Abstractions.Config;
using LJ.StreamSystem.LazyHare.Transport.Kafka;
using Microsoft.Extensions.Logging;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading.Tasks;

namespace LJ.StreamSystem.LazyHare.Internal;

public class AggregateStreamSystemSubscribe(IStreamSystemManager streamSystemManager,
        IStreamSystemMessageSerializer streamSystemMessageSerializer,
        IStreamSystemClientFactory streamSystemClientFactory,
        IAggregateStreamSystemMessageListener aggregateStreamSystemMessageListener,
        ILogger<AggregateStreamSystemSubscribe> logger) : IAggregateStreamSystemSubscribe
{
    private readonly IStreamSystemManager _streamSystemManager = streamSystemManager;
    private readonly IStreamSystemMessageSerializer _streamSystemMessageSerializer = streamSystemMessageSerializer;
    private readonly IStreamSystemClientFactory _streamSystemClientFactory = streamSystemClientFactory;
    private readonly IAggregateStreamSystemMessageListener _aggregateStreamSystemMessageListener = aggregateStreamSystemMessageListener;
    private readonly ILogger _logger= logger;


    public Task Start(CancellationToken stoppingToken)
    {
        // 加载所有的事件处理类
        var aggregateStreamSystemHandlerDescriptors = _streamSystemManager.GetAggregateStreamSystemHandlerDescriptors();

        // 注册事件订阅
        foreach (var aggregateStreamSystemHandlerDescriptor in aggregateStreamSystemHandlerDescriptors)
        {
            _ = Task.Factory.StartNew(async () =>
            {
                try
                {
                    // ReSharper disable once ConvertToUsingDeclaration
                    using var client = _streamSystemClientFactory.Create(aggregateStreamSystemHandlerDescriptor.ApplicationId, 
                        aggregateStreamSystemHandlerDescriptor.AggregateSummaryTopic,
                        aggregateStreamSystemHandlerDescriptor.AggregateTopics);

                    RegisterProcessor(client, stoppingToken);


                    await client.ListeningAsync( stoppingToken);
                }
                catch (OperationCanceledException)
                {
                    //ignore
                }
                catch (Exception e)
                {
                    _logger.LogError(e, "Exception:{Message}", e.Message);
                }
            }, stoppingToken, TaskCreationOptions.LongRunning, TaskScheduler.Default);
        }
        return Task.CompletedTask;
    }

    private void RegisterProcessor(IStreamSystemClient client, CancellationToken cancellationToken)
    {
        client.OnKeyCallback = async (transportData) =>
        {
            var applicationId = transportData.GetApplicationId();

            var aggregateStreamSystemHandlerDescriptor = _streamSystemManager.GetAggregateStreamSystemHandlerDescriptor(applicationId);

            var message = new AggregateStreamSystemMessage(transportData.Headers);
            message.Value = _streamSystemMessageSerializer.Deserialize(transportData.Body!, aggregateStreamSystemHandlerDescriptor.AggregateEventDataType);

            _logger.LogDebug("Key message. data:{data}", transportData);

            var data = await _aggregateStreamSystemMessageListener.OnKeyReceiveAsync(message, cancellationToken);

            return _streamSystemMessageSerializer.Serialize(data);
        };

        client.OnInitDataCallback = async (transportData) =>
        {
            var applicationId = transportData.GetApplicationId();

            var aggregateStreamSystemHandlerDescriptor = _streamSystemManager.GetAggregateStreamSystemHandlerDescriptor(applicationId);

            var message = new AggregateStreamSystemMessage(transportData.Headers);

            _logger.LogDebug("InitData. data:{data}", transportData);

            var data = await _aggregateStreamSystemMessageListener.OnInitReceiveAsync(message, cancellationToken);

            return _streamSystemMessageSerializer.Serialize(data);
        };

        client.OnSummaryDataCallback = async (transportData) =>
        {
            var applicationId = transportData.GetApplicationId();

            var aggregateStreamSystemHandlerDescriptor = _streamSystemManager.GetAggregateStreamSystemHandlerDescriptor(applicationId);

            var message = new AggregateStreamSystemMessage(transportData.Headers);

            _logger.LogDebug("InitData. data:{data}", transportData);

            var data = await _aggregateStreamSystemMessageListener.OnSummaryReceiveAsync(message, cancellationToken);

            return _streamSystemMessageSerializer.Serialize(data);
        };
    }

}
